package com.wdb007.venus.web.api.config;

import com.wdb.exchange.service.ApiService;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
import org.springframework.amqp.remoting.service.AmqpInvokerServiceExporter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class AmqpConfig {

    public static final String TOPIC_EXCHANGE    = "amq.topic";
    public static final String DIRECT_EXCHANGE    = "amq.direct";
    public static final String SERVER_QUEUE_NAME = ".ibms.server";
    public static final String WILL_MESSAGE_QUEUE_NAME   = ".ibms.will-message";
    public static final String DEADLETTER_QUEUE_NAME   = ".ibms.dead-letter";

    public static final String SERVICE_EXCHANGE       = ".ibms.exchange.service";
    public static final String API_SERVICE_QUEUE_NAME = ".ibms.service.api";
    public static final String API_RPC_QUEUE_NAME = ".ibms.rpc.api";

    @Bean
    String serverId() {
        return "server";
    }

    @Bean
    public AmqpAdmin amqpAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }
    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate template = new RabbitTemplate(connectionFactory);
        return template;
    }

    @Bean(name="aRabbitListenerContainerFactory")
    public SimpleRabbitListenerContainerFactory arabbitListenerContainerFactory(
            ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrentConsumers(3);
        return  factory;
    }

    @Bean(name="bRabbitListenerContainerFactory")
    public SimpleRabbitListenerContainerFactory brabbitListenerContainerFactory(
            ConnectionFactory connectionFactory) {
        SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
        factory.setConnectionFactory(connectionFactory);
        factory.setConcurrentConsumers(1);
        return  factory;
    }

    @Bean
    Queue queueForServer() {
        Map<String, Object> args = new HashMap<String, Object>();
        args.put("x-message-ttl", 60 * 1000);
        return QueueBuilder.durable(SERVER_QUEUE_NAME).withArguments(args).build();
    }
    @Bean
    Queue queueForWillMessage() {
        Map<String, Object> args = new HashMap<String, Object>();
        args.put("x-message-ttl", 3 * 60 * 1000);
        return QueueBuilder.durable(WILL_MESSAGE_QUEUE_NAME).withArguments(args).build();
    }
    @Bean
    Queue queueForDeadLetter() {
        Map<String, Object> args = new HashMap<String, Object>();
        args.put("x-message-ttl", 30 * 60 * 1000);
        return QueueBuilder.durable(DEADLETTER_QUEUE_NAME).withArguments(args).build();
    }
    @Bean
    Binding bindingForServer(Queue queueForServer) {
        return BindingBuilder.bind(queueForServer).to(new TopicExchange(TOPIC_EXCHANGE)).with(queueForServer.getName());
    }
    @Bean
    Binding bindingForWillMessage(Queue queueForWillMessage) {
        return BindingBuilder.bind(queueForWillMessage).to(new TopicExchange(TOPIC_EXCHANGE)).with(queueForWillMessage.getName());
    }
    @Bean
    Queue queueForService() {
        return QueueBuilder.durable(API_SERVICE_QUEUE_NAME).build();
    }
    @Bean
    Exchange exchangeForService() {
        return ExchangeBuilder.fanoutExchange(SERVICE_EXCHANGE).durable(true).build();
    }
    @Bean
    Binding bindingForService(Queue queueForService) {
        return BindingBuilder.bind(queueForService).to(new FanoutExchange(SERVICE_EXCHANGE));
    }

    @Autowired
    ApiService apiService;
    @Bean
    Queue queueForRpc() {
        Map<String, Object> args = new HashMap<String, Object>();
        args.put("x-message-ttl", 10 * 1000);
        return QueueBuilder.nonDurable(API_RPC_QUEUE_NAME).autoDelete().withArguments(args).build();
    }
    @Bean
    Binding bindingForRpc(Queue queueForRpc) {
        return BindingBuilder.bind(queueForRpc).to(new DirectExchange(DIRECT_EXCHANGE)).withQueueName();
    }
    @Bean
    AmqpInvokerServiceExporter exporter(ApiService apiService, AmqpTemplate template) {
        AmqpInvokerServiceExporter exporter = new AmqpInvokerServiceExporter();
        exporter.setServiceInterface(ApiService.class);
        exporter.setService(apiService);
        exporter.setAmqpTemplate(template);
//        exporter.setMessageConverter(new KryoMessageConverter());
        return exporter;
    }
    @Bean
    SimpleMessageListenerContainer listener(ConnectionFactory connectionFactory,
                                            AmqpInvokerServiceExporter exporter, Queue queueForRpc) {
        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer(connectionFactory);
        container.setMessageListener(exporter);
        container.setQueueNames(queueForRpc.getName());
        container.setConcurrentConsumers(3);
        return container;
    }
}
